package org.digitarts.rtms.spring.collectors.jms;

import java.util.Enumeration;
import java.util.HashSet;
import java.util.Set;

import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.jms.Session;

import org.digitarts.rtms.helpers.StringHelper;
import org.digitarts.rtms.spring.collectors.CollectorBase;

public class JMSBrowserCollector extends CollectorBase
{
	protected QueueConnectionFactory	queueConnectionFactory	= null;
	protected QueueConnection			queueConnection			= null;
	protected QueueSession				queueSession			= null;
	protected QueueBrowser				queueBrowser			= null;
	protected Set<Queue>				queues					= new HashSet<Queue>();

	@Override
	public void initCallback()
	{
		if (log.isDebugEnabled())
			log.debug("Initializing Queue Testing...");
		try
		{
			queueConnection = queueConnectionFactory.createQueueConnection();
			queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			// queueBrowser = queueSession.createBrowser(temporaryQueue);
			queueConnection.start();
		}
		catch (Exception e)
		{
			cleanupQueues();
			log.error("Failed to initialize queue testing", e);
		}
		finally
		{
		}
	}

	@SuppressWarnings("unused")
	@Override
	public void collectCallback() throws Exception
	{
		try
		{
			long start = System.currentTimeMillis();
			for (Queue queue : queues)
			{
				try
				{
					long innerStart = System.currentTimeMillis();
					int messageCount = 0;
					long oldestMessage = 0;
					queueBrowser = queueSession.createBrowser(queue);
					Enumeration<?> qEnum = queueBrowser.getEnumeration();
					while (qEnum.hasMoreElements())
					{
						messageCount++;
						Message message = (Message) qEnum.nextElement();
						long age = innerStart - message.getJMSTimestamp();
						if (age > oldestMessage)
						{
							oldestMessage = age;
						}
					}
					long innerElapsed = System.currentTimeMillis() - innerStart;
					tracer.traceSticky(messageCount, StringHelper.append(tracingNameSpace, queue.getQueueName(), "Message Count"));
					if (oldestMessage > 0)
					{
						tracer.traceSticky(oldestMessage, StringHelper.append(tracingNameSpace, queue.getQueueName(), "Age of Oldest Message"));
					}
				}
				catch (Exception e)
				{
					if (logErrors)
						log.error("JMSBrowserCollector failed on queue:[" + queue + "]", e);
				}
				finally
				{
					try
					{
						queueBrowser.close();
					}
					catch (Exception e)
					{
					}
				}
			}
			long elapsed = System.currentTimeMillis() - start;
			tracer.trace(elapsed, elapsedTimeMetricNameSpace);
		}
		catch (Exception e)
		{
			if (logErrors)
			{
				log.error("JMSBrowserCollector Failure", e);
			}
		}
	}

	@Override
	public void stopCallback()
	{
		cleanupQueues();
	}

	protected void cleanupQueues()
	{
		try
		{
			queueSession.close();
		}
		catch (Exception e)
		{
		}
		try
		{
			queueConnection.stop();
		}
		catch (Exception e)
		{
		}
		try
		{
			queueConnection.close();
		}
		catch (Exception e)
		{
		}
	}

	public Set<Queue> getQueues()
	{
		return queues;
	}

	public void setQueues(Set<Queue> queues)
	{
		this.queues = queues;
	}

	public void setQueueConnectionFactory(QueueConnectionFactory queueConnectionFactory)
	{
		this.queueConnectionFactory = queueConnectionFactory;
	}
}
